package com.zhang.sparkstreaming_2

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Duration, StreamingContext}

import java.sql.{Connection, PreparedStatement, ResultSet}
import java.text.SimpleDateFormat
import java.util.Date
import scala.collection.mutable.ListBuffer

/**
 * @title: 广告黑名单
 * @author: zhang
 * @date: 2022/2/21 11:12
 */
object SparkStreaming08_req1 {

  def main(args: Array[String]): Unit = {
    //todo 获取环境
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("spark streaming")
    val ssc = new StreamingContext(conf, Duration(5 * 1000L))

    //todo kafka参数定义
    val kafkaPara: Map[String, String] = Map[String, String](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "zhang",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )


    // todo 从kafka读取数据
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Set("spark_test"), kafkaPara))

    val kafkaDS: DStream[Ads_log] = kafkaDStream.map(
      kafkaStr => {
        val fields: Array[String] = kafkaStr.value().split(" ")
        val sdf = new SimpleDateFormat("yyyy-MM-dd")
        Ads_log(sdf.format(new Date(fields(0).toLong)), fields(1), fields(2), fields(3), fields(4))
      }
    )

    //todo step1-----
    val dayResult: DStream[((String, String, String), Int)] = kafkaDS.transform(
      rdd => {
        //todo 周期性获取黑名单数据
        val list = ListBuffer[String]()
        val conn: Connection = JDBCUtil.getConnection
        val statement: PreparedStatement = conn.prepareStatement(
          """
            |select * from black_list
            |""".stripMargin)
        val resultSet: ResultSet = statement.executeQuery()
        while (resultSet.next()) {
          list.append(resultSet.getString(1))
        }

        //todo 判断点击用户是否在黑名单
        val filterRDD: RDD[Ads_log] = rdd.filter(
          data => {
            !list.contains(data.user)
          }
        )

        //todo 如果用户不在黑名单中，那么进行统计数量（每个采集周期）
        filterRDD.map(
          data => {
            ((data.time, data.user, data.adid), 1)
          }
        ) reduceByKey (_ + _)

      }
    )


    //todo step2-----
    dayResult.foreachRDD(
      rdd => {
        rdd.foreachPartition(
          iter => {
            val conn = JDBCUtil.getConnection
             //INSERT INTO black_list (userid) VALUES (?) ON DUPLICATE KEY update userid=?
            val insert: PreparedStatement = conn.prepareStatement(
              """
                |insert into black_list
                | (userid)
                | values (?)
                | ON DUPLICATE KEY
                | UPDATE userid = ?
                |""".stripMargin)

            val upsert: PreparedStatement = conn.prepareStatement(
              """
                |insert into user_ad_count
                | (dt,userid,adid,count)
                | values(?, ?, ?, ?)
                | ON DUPLICATE KEY
                | UPDATE count = count + ?
                |""".stripMargin)


            iter.foreach {
              case ((day, user, ad), cnt) => {

                if (cnt >= 10) {
                  //todo 如果统计数量超过点击阈值，那么将用户拉入黑名单
                  val insert: PreparedStatement = conn.prepareStatement(
                    """
                      |insert into black_list
                      | (userid)
                      | values (?)
                      | ON DUPLICATE KEY
                      | UPDATE userid = ?
                      |""".stripMargin)
                  insert.setString(1, user)
                  insert.setString(2, user)
                  insert.execute()
                } else {
                  //todo 如果没有超过阈值，那么需要将当天的广告点击数量进行更新
                  upsert.setString(1, day)
                  upsert.setString(2, user)
                  upsert.setString(3, ad)
                  upsert.setInt(4, cnt)
                  upsert.setInt(5, cnt)
                  upsert.execute()
                  //todo 判断更新后的点击数据是否超过阈值，如果超过，那么将用户拉入到黑名单。
                  val select: PreparedStatement = conn.prepareStatement(
                    """
                      |select * from user_ad_count
                      | where count >= 10
                      |""".stripMargin)
                  val resultSet: ResultSet = select.executeQuery()
                  if (resultSet.next()){
                    insert.setString(1, user)
                    insert.setString(2, user)
                    insert.execute()
                    select.close()
                  }
                }
              }
            }
            conn.close()
            insert.close()
            upsert.close()
          }
        )
      }
    )

    //启动采集器
    ssc.start()
    ssc.awaitTermination()
  }

  case class Ads_log(time: String,
                     area: String,
                     city: String,
                     user: String,
                     adid: String)
}
